package com.project.mq.mqserver.core;

import com.project.mq.common.Consumer;
import com.project.mq.common.ConsumerEnv;
import com.project.mq.common.MqException;
import com.project.mq.mqserver.VirtualHost;
import lombok.Data;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 通过这个类.来实现消费消息的核心逻辑
 */

@Data
public class ConsumerManager {
    //持有上层VirtualHost对象的引用,用来操作数据
    private VirtualHost parent;
    // 存放令牌的队列。通过令牌来触发消费线程的消费操作。
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //指定一个线程池,负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //扫描线程
    private Thread scannerThread = null;

    //扫描线程
    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;

        // 启动扫描线程
        Thread scanThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 找到队列
                    MSGQueue msgQueue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (msgQueue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后,队列不存在! queueName=" + queueName);
                    }
                    // 3. 消费一个数据
                    synchronized (msgQueue) {
                        consumeMessage(msgQueue);
                    }
                } catch (MqException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "scanThread");
        //设置这个线程为后台线程
        scanThread.setDaemon(true);
        scanThread.start();
    }

    // 通知消费者去消费消息
    //这个方法的调用时机就h是发送消息的时候
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对印的队列
        MSGQueue msgQueue = parent.getMemoryDataCenter().getQueue(queueName);
        if (msgQueue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (msgQueue) {
            msgQueue.addConsumerEnv(consumerEnv);

            //// 如果当前队列中已经有了一些消息了，需要立即就消费掉。
            // 把已经积压的 n 个数据都先消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                //调用一次,就消费一条消息
                consumeMessage(msgQueue);
            }
        }
    }

    //消费消息
    private void consumeMessage(MSGQueue msgQueue) throws MqException {
        // 1. 按照轮询方式，先找个消费者出来
        ConsumerEnv luckyDog = msgQueue.chooseConsumer();
        if (luckyDog == null) {
            // 如果当前还没有订阅者，就先暂时不消费。
            return;
        }

        // 2. 从指定队列中取一个元素
        Message message = parent.getMemoryDataCenter().pollMessage(msgQueue.getName());
        if (message == null) {
            return;
        }


        // 3. 丢到线程池中干活，回调执行时间可能比较长，不适合让扫描线程去调用。
        workerPool.submit(() -> {
            try {
                // 1. 先把消息放到待确认队列中
                // （这个逻辑必须放到执行回调前面，如果是 autoAck false，在回调内部会调用 basicAck，执行彻底删除需要先放到待确认队列，才能彻底删除）
                parent.getMemoryDataCenter().addMessageWaitAck(msgQueue.getName(), message);

                // 2. 调用消费者的回调，如果回调抛出异常了，则不会对消息进行任何 ack 操作，相当于消息仍然处在待消费的状态。
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());

                // 3. 如果消息是自动确认，则可以直接把消息彻底删除了。
                // （这个逻辑必须放到执行回调后面，万一执行回调一半服务器崩溃，这个消息仍然存在于硬盘上，下次启动还可以被继续消费到）
                //如果当前是 “自动应答”，就可以直接把消息删除了。
                //如果当前是 “手动应答”，则先不处理，交给后续消费者调用 basicAck 方法来处理。
                if (luckyDog.isAutoAck()) {
                    // 则修改硬盘上的消息为"无效"，同时删除内存中的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(msgQueue, message);
                    }
                    //删除待确认集合的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(msgQueue.getName(), message.getMessageId());
                    //删除消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + msgQueue.getName() + ", messageId=" + message.getMessageId());
                }
            } catch (MqException | IOException | ClassNotFoundException e) {
                e.printStackTrace();
            }
        });
    }



}